package jarvis.web;

import jarvis.utils.ESClientFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.spark.api.java.JavaEsSpark;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

@Controller
public class DuplicateController implements Serializable {
	private static final long serialVersionUID = 1L;

	private transient Client esclient = ESClientFactory.getClient();

	private static class KeyText implements Serializable {
		String key;
		String text;
	}

	@RequestMapping("/duplicates")
	@ResponseBody
	public String findDuplicates() {
		SparkConf sparkConf = new SparkConf().setAppName("JavaSparkPi")
				.setMaster("local");
		JavaSparkContext jsc = new JavaSparkContext(sparkConf);
		JavaRDD<Map<String, Object>> esRDD = JavaEsSpark.esRDD(jsc,
				"jarvis/filedata");

		final List<List<String>> dupKeysList = new ArrayList<List<String>>();
		Map<String, List<KeyText>> hashedLists = esRDD
				.map(new Function<Map<String, Object>, Map<String, List<KeyText>>>() {

					public Map<String, List<KeyText>> call(
							Map<String, Object> arg0) throws Exception {
						Map<String, List<KeyText>> hashed = new HashMap<String, List<KeyText>>();
						for (String k : arg0.keySet()) {
							Object o = arg0.get(k);
							if (o instanceof Map) {
								Map m = (Map) o;
								Object text = m.get("text");
								if (text != null) {
									String hash = String.valueOf(text
											.hashCode());
									KeyText kt = new KeyText();
									kt.key = k;
									kt.text = text.toString();
									if (!hashed.containsKey(hash)) {
										hashed.put(hash,
												new ArrayList<KeyText>());
									}
									hashed.get(hash).add(kt);
								}
							}
						}
						return hashed;
					}

				})
				.reduce(new Function2<Map<String, List<KeyText>>, Map<String, List<KeyText>>, Map<String, List<KeyText>>>() {

					public Map<String, List<KeyText>> call(
							Map<String, List<KeyText>> m0,
							Map<String, List<KeyText>> m1) throws Exception {
						Map<String, List<KeyText>> arg0 = new ConcurrentHashMap<String, List<KeyText>>();
						arg0.putAll(m0);
						Map<String, List<KeyText>> arg1 = new ConcurrentHashMap<String, List<KeyText>>();
						arg1.putAll(m1);
						for (String k : arg1.keySet()) {
							if (!arg0.containsKey(k)) {
								arg0.put(
										k,
										new ArrayList<DuplicateController.KeyText>());
							}
							arg0.get(k).addAll(arg1.get(k));
						}
						return arg0;
					}
				});
		for (String k : hashedLists.keySet()) {
			if (hashedLists.get(k).size() < 2) {
				hashedLists.remove(k);
			} else {
				List<KeyText> l = hashedLists.get(k);
				Collections.sort(l, new Comparator<KeyText>() {

					public int compare(KeyText o1, KeyText o2) {
						return o1.text.compareTo(o2.text);
					}
				});
				int i = 0;
				while (i < l.size()) {
					List<KeyText> dups = new ArrayList<KeyText>();
					dups.add(l.get(i));
					i++;
					while (i < l.size()
							&& l.get(i).text.equals(dups
									.get(dups.size() - 1).text)) {
						dups.add(l.get(i));
						i++;
					}
					if (dups.size() > 1) {
						List<String> dupKeys = new ArrayList<String>();
						for (KeyText kt : dups) {
							dupKeys.add(kt.key);
						}
						dupKeysList.add(dupKeys);
					}
				}
			}
		}
		List<SearchResponse> srs = new ArrayList<SearchResponse>();
		for (List<String> keys : dupKeysList) {
			SearchResponse response = esclient.prepareSearch("jarvis")
					.setTypes("filedata")
					.setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
					.setQuery(QueryBuilders.termsQuery("_id", keys)).setFrom(0)
					.setSize(60).setExplain(true).execute().actionGet();
			srs.add(response);
		}
		return srs.toString();
	}

	public static void main(String[] args) {
		new DuplicateController().findDuplicates();
	}
}
